package cn.com.sparkle.paxos.deamon;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.CountDownLatch;

import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.AccountBook;
import cn.com.sparkle.paxos.ClusterState;
import cn.com.sparkle.paxos.Configuration;
import cn.com.sparkle.paxos.NodesCollection;
import cn.com.sparkle.paxos.event.CatchUpEvent;
import cn.com.sparkle.paxos.event.EventsManager;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.CatchUpRecord;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.CatchUpRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.CatchUpResponse;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.MessagePackage;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.SuccessfulRecord;
import cn.com.sparkle.paxos.net.CallBack;
import cn.com.sparkle.paxos.net.NetNode;
//import cn.com.sparkle.paxos.net.message.CatchUpRequest;
//import cn.com.sparkle.paxos.net.message.CatchUpResponse;
//import cn.com.sparkle.paxos.net.protobuffmessage.PaxosMessages.CatchUpRequest;
//import cn.com.sparkle.paxos.stablestorage.model.SuccessfulRecord;

public class CatchUpDeamon implements Runnable {
	private final Logger logger = Logger.getLogger(CatchUpDeamon.class);

	private final static long BACKWARD_TOLERATION_TIME = 15 * 1000;

	private ClusterState cState;
	// private Configuration conf;
	private AccountBook aBook;
	private EventsManager eventsManager;
	private String selfAddress;

	public CatchUpDeamon(EventsManager eventsManager, ClusterState cState,
			Configuration conf, AccountBook aBook) {
		this.eventsManager = eventsManager;
		this.cState = cState;
		this.selfAddress = conf.getSelfAddress();
		this.aBook = aBook;
	}

	@Override
	public void run() {
		int curCatchUpState = CatchUpEvent.FAIL_CATCH_UP;
		long lastCanExecuteInstanceIdRecord = -1;
		long lastInstanceIdCheckTime = 0;
		while (true) {
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
			}
			NodesCollection senators = cState.getSenators();
			List<NetNode> activeSenator = senators.getOutOfOrderList();
			long lastCanExecuteInstanceId = aBook
					.getLastCanExecutableInstanceId();
			logger.debug("lastCanExecuteInstanceId:" + lastCanExecuteInstanceId
					+ " lastCanExecuteInstanceIdRecord:"
					+ lastCanExecuteInstanceIdRecord + " waitInterval:"
					+ (System.currentTimeMillis() - lastInstanceIdCheckTime));
			if (lastCanExecuteInstanceId != lastCanExecuteInstanceIdRecord) {
				lastCanExecuteInstanceIdRecord = lastCanExecuteInstanceId;
				lastInstanceIdCheckTime = System.currentTimeMillis();
			} else if (System.currentTimeMillis() - lastInstanceIdCheckTime > BACKWARD_TOLERATION_TIME) {
				// find max instance id from cluster
				long maxInstanceId = -1;
				int validCount = 0;// the count of valid senator that's node
									// state is assigned by heart beat
				for (NetNode nnode : activeSenator) {
					if (nnode.getNodeState().isConnected()
							&& !nnode.getNodeState().getAddress()
									.equals(selfAddress)) {
						if (nnode.getNodeState().getLastCanExecuteInstanceId() > maxInstanceId) {
							maxInstanceId = nnode.getNodeState()
									.getLastCanExecuteInstanceId();
						}
						++validCount;
					}
				}
				long selfInstanceId = aBook.getLastCanExecutableInstanceId();
				long tempselfInstanceId = -1;
				logger.debug("before catchUp state( maxInstanceId:"
						+ maxInstanceId + " selfInstanceId:" + selfInstanceId
						+ ")");
				while (maxInstanceId > selfInstanceId) {

					tempselfInstanceId = selfInstanceId;
					for (NetNode nnode : activeSenator) {
						if (nnode.getNodeState().getAddress()
								.equals(selfAddress)) {
							continue;
						}
						long nnodeInstanceId = nnode.getNodeState()
								.getLastCanExecuteInstanceId();
						nnodeInstanceId = maxInstanceId > nnodeInstanceId ? nnodeInstanceId
								: maxInstanceId;
						// try study newer instance than self's
						while (nnodeInstanceId > selfInstanceId) {
							logger.debug("catch up from " + nnode.getNodeState().getAddress());
							long catchUpInstanceId = aBook
									.getFisrtInstnaceIdOfWaitInsertToExecuteQueue();
							if (catchUpInstanceId == -1
									|| catchUpInstanceId > nnodeInstanceId) {
								catchUpInstanceId = nnodeInstanceId;
							}
							// start process of catch up
							catchUp(nnode, selfInstanceId + 1,
									(int) (catchUpInstanceId - selfInstanceId));
							selfInstanceId = aBook
									.getLastCanExecutableInstanceId();
							if (selfInstanceId < catchUpInstanceId) {
								break;
							}
						}
					}
					// if the selfInstanceId is not changed, it will can't study
					// instance from others
					if (tempselfInstanceId == selfInstanceId) {
						break;
					}
				}
				logger.debug("after catchUp state( maxInstanceId:"
						+ maxInstanceId + " selfInstanceId:" + selfInstanceId
						+ " validCount:" + validCount + " activeNodeSize:"
						+ activeSenator.size() + ")");

				int quorum = senators.getNodeMembers().size() / 2;// except self
				if (validCount >= quorum) {// valid from more than quorum ,this
											// is the condition to trigger
											// events.
					if (maxInstanceId > selfInstanceId) {
						// fail to catch up,activate failure method
						if (curCatchUpState == CatchUpEvent.RECOVERY_FROM_CATCH_UP) {
							CatchUpEvent.doCatchUpFailEvent(eventsManager);
							logger.debug("catch up fail");
						}
					} else {
						if (curCatchUpState == CatchUpEvent.FAIL_CATCH_UP) {
							CatchUpEvent.doRecoveryFromFailEvent(eventsManager);
							logger.debug("catch up success");
						}
					}
				}
				lastInstanceIdCheckTime = System.currentTimeMillis();
			}
		}
	}

	private void catchUp(NetNode n, long startInstanceId, final int size) {
		final CountDownLatch countDownLatch = new CountDownLatch(1);
		CatchUpRequest.Builder builder = CatchUpRequest.newBuilder();
		builder.setStartInstanceId(startInstanceId).setSize(size);

		n.write(builder.build(), new CallBack() {
			@Override
			public void fail(NetNode nnode) {
				countDownLatch.countDown();
			}

			@Override
			public void call(NetNode nnode, MessagePackage value) {

				CatchUpResponse response = value.getCatchUpResponse();
				logger.debug("study " + response.getSuccessfulRecordsCount() + " successful records");
				if (response.getSuccessfulRecordsCount() != 0) {
					for (CatchUpRecord r : response.getSuccessfulRecordsList()) {
						try {
							aBook.writeSuccessfulRecord(r.getInstanceId(),
									r.getRecord(), null);
						} catch (IOException e) {
							logger.error("fatal error:", e);
						}
					}
				}
				if (value.getIsLast()) {
					countDownLatch.countDown();
				}
			}
		});
		try {
			countDownLatch.await();
		} catch (InterruptedException e) {
		}
	}

}
